package com.boarsoft.elasticsearch;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.get.MultiGetItemResponse;
import org.elasticsearch.action.get.MultiGetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.script.Script;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ElasticSearchClientImpl implements ElasticSearchClient {
	private static final Logger log = LoggerFactory.getLogger(ElasticSearchClientImpl.class);

	protected String serverName;// es服务器地址
	protected int port;// es服务器端口
	protected String clusterName;// es服务器名称

	/** 非注入，内部变量 */
	protected TransportClient client;

	@PostConstruct
	public void init() throws UnknownHostException, InterruptedException, ExecutionException {
		Settings ess = Settings.builder().put("cluster.name", clusterName)//
				.put("client.transport.sniff", true).build();
		client = new PreBuiltTransportClient(ess);
		client.addTransportAddress(new TransportAddress(InetAddress.getByName(serverName), port));
	}

	@Override
	public TransportClient getClient() {
		return client;
	}

	@Override
	public String get(String index, String type, String id) {
		GetResponse gr = client.prepareGet(index, type, id).execute().actionGet();
		return gr.getSourceAsString();
	}

	@Override
	public String save(String index, String type, String id, Map<String, String> m, long timeout, TimeUnit timeUnit) {
		IndexRequestBuilder irb = client.prepareIndex(index, type, id);
		irb.setTimeout(new TimeValue(timeout, timeUnit)); // ms
		IndexResponse ir = irb.setSource(m, XContentType.JSON).get();
		//
		// IndexResponse ir = irb.setSource(m).execute().actionGet();
		//
		// ObjectMapper om = new ObjectMapper();
		// String json = om.writeValueAsString(object);
		// ActionFuture<IndexResponse> af = client.index(new IndexRequest(index,
		// type).source(json));
		return ir.getIndex();
	}

	@Override
	public void bulkSave(String index, String type, List<Map<String, Object>> list, long timeout, TimeUnit timeUnit)
			throws Exception {
		BulkRequestBuilder brb = client.prepareBulk();
		for (Map<String, Object> m : list) {
			brb.add(client.prepareIndex(index, type, null).setSource(m, XContentType.JSON));
		}
		BulkResponse br = brb.get();
		log.info("Bulk save {} records to index {} in {}ms", list.size(), index, br.getTook().millis());
	}

	@Override
	public void truncate(String index) throws InterruptedException, ExecutionException {
		client.delete(new DeleteRequest(index)).get();
	}

	@Override
	public void delete(String index, String id) throws InterruptedException, ExecutionException {
		client.delete(new DeleteRequest(index, "none", id)).get();
	}

	// -------------------

	/**
	 * 查看集群信息
	 */
	public void testInfo() {
		List<DiscoveryNode> nodes = client.connectedNodes();
		for (DiscoveryNode dn : nodes) {
			System.out.println(dn.getHostAddress());
		}
	}

	public void update() throws Exception {
		UpdateRequest ur = new UpdateRequest();
		ur.index("twitter");
		ur.type("tweet");
		ur.id("1");
		ur.doc(XContentFactory.jsonBuilder().startObject()
				// 对没有的字段添加, 对已有的字段替换
				.field("gender", "male").field("message", "hello").endObject());
		UpdateResponse response = client.update(ur).get();

		// 打印
		String index = response.getIndex();
		String type = response.getType();
		String id = response.getId();
		long version = response.getVersion();
		System.out.println(index + " : " + type + ": " + id + ": " + version);
	}

	/**
	 * 测试update api, 使用client
	 * 
	 * @throws Exception
	 */
	public void testUpdate2() throws Exception {
		// 使用Script对象进行更新
		// UpdateResponse response = client.prepareUpdate("twitter", "tweet",
		// "1")
		// .setScript(new Script("hits._source.gender = \"male\""))
		// .get();

		// 使用XContFactory.jsonBuilder() 进行更新
		// UpdateResponse response = client.prepareUpdate("twitter", "tweet",
		// "1")
		// .setDoc(XContentFactory.jsonBuilder()
		// .startObject()
		// .field("gender", "malelelele")
		// .endObject()).get();

		// 使用ur对象及script
		// UpdateRequest ur = new UpdateRequest("twitter", "tweet",
		// "1")
		// .script(new Script("ctx._source.gender=\"male\""));
		// UpdateResponse response = client.update(ur).get();

		// 使用ur对象及documents进行更新
		UpdateResponse response = client.update(new UpdateRequest("twitter", "tweet", "1")
				.doc(XContentFactory.jsonBuilder().startObject().field("gender", "male").endObject())).get();
		System.out.println(response.getIndex());
	}

	public void testUpdate3() throws InterruptedException, Exception {
		UpdateRequest rq = new UpdateRequest("twitter", "tweet", "1").script(new Script("ctx._source.gender=\"male\""));
		UpdateResponse rp = client.update(rq).get();
		System.out.println(rp);
	}

	public void update2() throws Exception {
		// 设置查询条件, 查找不到则添加生效
		IndexRequest ir = new IndexRequest("twitter", "tweet", "2")
				.source(XContentFactory.jsonBuilder().startObject().field("name", "214").field("gender", "gfrerq").endObject());
		// 设置更新, 查找到更新下面的设置
		UpdateRequest ur = new UpdateRequest("twitter", "tweet", "2")
				.doc(XContentFactory.jsonBuilder().startObject().field("user", "wenbronk").endObject()).upsert(ir);
		client.update(ur).get();
	}

	public void testMultiGet() {
		MultiGetResponse multiGetResponse = client.prepareMultiGet().add("twitter", "tweet", "1")
				.add("twitter", "tweet", "2", "3", "4").add("anothoer", "type", "foo").get();

		for (MultiGetItemResponse itemResponse : multiGetResponse) {
			GetResponse response = itemResponse.getResponse();
			if (response.isExists()) {
				String sourceAsString = response.getSourceAsString();
				System.out.println(sourceAsString);
			}
		}
	}

	// public void testBulkProcessor() throws Exception {
	// // 创建BulkPorcessor对象
	// BulkProcessor bulkProcessor = BulkProcessor.builder(client, new
	// Listener() {
	// public void beforeBulk(long paramLong, BulkRequest paramBulkRequest) {
	// // TODO Auto-generated method stub
	// }
	//
	// // 执行出错时执行
	// public void afterBulk(long paramLong, BulkRequest paramBulkRequest,
	// Throwable paramThrowable) {
	// // TODO Auto-generated method stub
	// }
	//
	// public void afterBulk(long paramLong, BulkRequest paramBulkRequest,
	// BulkResponse paramBulkResponse) {
	// // TODO Auto-generated method stub
	// }
	// })
	// // 1w次请求执行一次bulk
	// .setBulkActions(10000)
	// // 1gb的数据刷新一次bulk
	// .setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
	// // 固定5s必须刷新一次
	// .setFlushInterval(TimeValue.timeValueSeconds(5))
	// // 并发请求数量, 0不并发, 1并发允许执行
	// .setConcurrentRequests(1)
	// // 设置退避, 100ms后执行, 最大请求3次
	// .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100),
	// 3)).build();
	//
	// // 添加单次请求
	// bulkProcessor.add(new IndexRequest("twitter", "tweet", "1"));
	// bulkProcessor.add(new DeleteRequest("twitter", "tweet", "2"));
	//
	// // 关闭
	// bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
	// // 或者
	// bulkProcessor.close();
	// }

	/**
	 * 测试查询
	 */
	public void testSearch() {
		// SearchRequestBuilder searchRequestBuilder =
		// client.prepareSearch("twitter", "tweet", "1");
		// SearchResponse response = searchRequestBuilder.setTypes("type1",
		// "type2")
		// .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
		// .setQuery(QueryBuilders.termQuery("user", "test"))
		// .setPostFilter(QueryBuilders.rangeQuery("age").from(0).to(1))
		// .setFrom(0).setSize(2).setExplain(true)
		// .execute().actionGet();
		SearchResponse response = client.prepareSearch().execute().actionGet();
		// SearchHits hits = response.getHits();
		// for (SearchHit searchHit : hits) {
		// for(Iterator<SearchHitField> iterator = searchHit.iterator();
		// iterator.hasNext(); ) {
		// SearchHitField next = iterator.next();
		// System.out.println(next.getValues());
		// }
		// }
		System.out.println(response);
	}

	/**
	 * 测试scroll api 对大量数据的处理更有效
	 */
	public void testScrolls() {
		// QueryBuilder queryBuilder = QueryBuilders.termQuery("twitter",
		// "tweet");

		// SearchResponse response =
		// client.prepareSearch("twitter").addSort(SortParseElement.DOC_FIELD_NAME,
		// SortOrder.ASC)
		// .setScroll(new
		// TimeValue(60000)).setQuery(queryBuilder).setSize(100).execute().actionGet();
		//
		// while (true) {
		// for (SearchHit hit : response.getHits().getHits()) {
		// System.out.println("i am coming");
		// }
		// SearchResponse response2 =
		// client.prepareSearchScroll(response.getScrollId()).setScroll(new
		// TimeValue(60000))
		// .execute().actionGet();
		// if (response2.getHits().getHits().length == 0) {
		// System.out.println("oh no=====");
		// break;
		// }
		// }

	}

	/**
	 * 测试multiSearch
	 */
	// public void testMultiSearch() {
	// QueryBuilder qb1 = QueryBuilders.queryStringQuery("elasticsearch");
	// SearchRequestBuilder requestBuilder1 =
	// client.prepareSearch().setQuery(qb1).setSize(1);
	//
	// QueryBuilder qb2 = QueryBuilders.matchQuery("user", "kimchy");
	// SearchRequestBuilder requestBuilder2 =
	// client.prepareSearch().setQuery(qb2).setSize(1);
	//
	// MultiSearchResponse multiResponse =
	// client.prepareMultiSearch().add(requestBuilder1).add(requestBuilder2).execute()
	// .actionGet();
	// long nbHits = 0;
	// for (MultiSearchResponse.Item item : multiResponse.getResponses()) {
	// SearchResponse response = item.getResponse();
	// nbHits = response.getHits().getTotalHits();
	// SearchHit[] hits = response.getHits().getHits();
	// System.out.println(nbHits);
	// }
	//
	// }

	/**
	 * 测试聚合查询
	 */
	public void testAggregation() {
		// SearchResponse response =
		// client.prepareSearch().setQuery(QueryBuilders.matchAllQuery()) //
		// 先使用query过滤掉一部分
		// .addAggregation(AggregationBuilders.terms("term").field("user"))
		// .addAggregation(AggregationBuilders.dateHistogram("agg2").field("birth").interval(DateHistogramInterval.YEAR))
		// .execute().actionGet();
		// Aggregation aggregation2 = response.getAggregations().get("term");
		// Aggregation aggregation = response.getAggregations().get("agg2");
		// SearchResponse response2 = client.search(new
		// SearchRequest().searchType(SearchType.QUERY_AND_FETCH)).actionGet();
	}

	/**
	 * 测试terminate
	 */
	public void testTerminateAfter() {
		SearchResponse response = client.prepareSearch("twitter").setTerminateAfter(1000).get();
		if (response.isTerminatedEarly()) {
			System.out.println("ternimate");
		}
	}

	/**
	 * 过滤查询: 大于gt, 小于lt, 小于等于lte, 大于等于gte
	 */
	// public void testFilter() {
	// SearchResponse response =
	// client.prepareSearch("twitter").setTypes("").setQuery(QueryBuilders.matchAllQuery())
	// // 查询所有
	// .setSearchType(SearchType.QUERY_THEN_FETCH)
	// // .setPostFilter(FilterBuilders.rangeFilter("age").from(0).to(19)
	// // .includeLower(true).includeUpper(true))
	// // .setPostFilter(FilterBuilderFactory
	// // .rangeFilter("age").gte(18).lte(22))
	// .setExplain(true) // explain为true表示根据数据相关度排序，和关键字匹配最高的排在前面
	// .get();
	// }

	/**
	 * 分组查询
	 */
	// public void testGroupBy() {
	// // 根据user进行分组
	// client.prepareSearch("twitter").setTypes("tweet").setQuery(QueryBuilders.matchAllQuery())
	// .setSearchType(SearchType.QUERY_THEN_FETCH)
	// .addAggregation(AggregationBuilders.terms("user").field("user").size(0)).get();
	// }

	// // 获取注入的参数
	// public void setIndexInfos(Map<String, Pair<String, String>> indexInfos) {
	// this.indexInfos = indexInfos;
	// }
	//
	// public Map<String, Pair<String, String>> getIndexInfos() {
	// return this.indexInfos;
	// }
	//
	// // 获取index
	// private String getIndex(Class klass) {
	// return indexInfos.get(klass.getName()).fst;
	// }
	//
	// // 获取type
	// private String getType(Class klass) {
	// return indexInfos.get(klass.getName()).snd;
	// }

	public String getServerName() {
		return serverName;
	}

	public void setServerName(String serverName) {
		this.serverName = serverName;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public String getClusterName() {
		return clusterName;
	}

	public void setClusterName(String clusterName) {
		this.clusterName = clusterName;
	}
}